package com.boarsoft.message.bean;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;

import javax.annotation.PostConstruct;

import com.boarsoft.message.store.MessageRoller;

/**
 * 用于缓冲消息的队列，当队列容量为0时，作为配置类使用<br>
 * 队列本身不提供数据保存的配置和实现，而是通过roller来完成<br>
 * 数据是写文件还是记到库，是同步还是异步刷盘，均取决于roller
 * 
 * @author Mac_J
 *
 */
public abstract class MessageQueue {
	/** 全局唯一编号，用于唯一标识此队列 */
	protected String code;
	/** 此队列归属哪个系统，可选配置项，方便管理 */
	protected String catalog;
	/** 队列中文名，可选配置项，方便管理 */
	protected String name;
	/** 队列优先级（一次发送几个） */
	protected int priority = 50;
	/** 队列最大长度 */
	protected int capacity = 10000;

	/** 为空表示纯内存队列 */
	protected MessageRoller roller;
	/** */
	protected BlockingQueue<MessageEntity> queue;

	/** 大致的数量？ */
	protected AtomicLong size = new AtomicLong(0L);
	/** 大致的数量？ */
	protected AtomicLong in = new AtomicLong(0L);
	/** 大致的数量？ */
	protected AtomicLong out = new AtomicLong(0L);

	@PostConstruct
	public void init() {
		if (queue == null) {
			queue = new LinkedBlockingQueue<MessageEntity>(capacity);
		}
	}

	public String toString() {
		return code;
	}

	/**
	 * @return 可能不准确的队列大小
	 */
	public int getSize() {
		return new Long(size.get()).intValue();
	}

	/**
	 * @return 可能不准确的队列大小
	 */
	public int getIn() {
		return new Long(in.getAndSet(0L)).intValue();
	}

	/**
	 * @return 可能不准确的队列大小
	 */
	public int getOut() {
		return new Long(out.getAndSet(0L)).intValue();
	}

	/**
	 * 此方法不会将消息持久化，适用于从存储中恢复消息队列，或不需要持久化的场景
	 * 
	 * @param m
	 * @return
	 * @throws InterruptedException 
	 * @throws Exception
	 */
	public boolean add(Message m) throws InterruptedException {
		MessageEntity me = new MessageEntity(m, roller);
		return this.add(me);
	}

	/**
	 * 将消息添加到队列，如果有配置roller就调用它完成持久化
	 * 
	 * @param m
	 * @return
	 * @throws Exception
	 */
	public boolean put(Message m) throws Exception {
		MessageEntity me = new MessageEntity(m, roller);
		me.persist();
		return this.add(me);
	}

	/**
	 * 直接添加已持久化的消息，避免再次持久化
	 * 
	 * @param me
	 * @return
	 * @throws InterruptedException 
	 */
	public boolean add(MessageEntity me) throws InterruptedException {
		queue.put(me);
		in.incrementAndGet();
		size.incrementAndGet();
		return true;
	}

	public MessageEntity take() throws InterruptedException {
		MessageEntity me = queue.take();
		out.incrementAndGet();
		size.decrementAndGet();
		return me;
	}

	public MessageEntity poll() {
		MessageEntity me = queue.poll();
		if (me == null) {
			return null;
		}
		out.incrementAndGet();
		size.decrementAndGet();
		return me;
	}

	public boolean isEmpty() {
		return queue == null || queue.isEmpty();
	}

	public int getPriority() {
		return priority;
	}

	public void setPriority(int priority) {
		if (priority < 1) {
			throw new IllegalArgumentException("Priority of message queue must > 1");
		}
		this.priority = priority;
	}

	public String getCode() {
		return code;
	}

	public void setCode(String code) {
		this.code = code;
	}

	public MessageRoller getRoller() {
		return roller;
	}

	public void setRoller(MessageRoller roller) {
		this.roller = roller;
	}

	public int getCapacity() {
		return capacity;
	}

	public void setCapacity(int capacity) {
		this.capacity = capacity;
	}

	public int size() {
		return queue.size();
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public String getCatalog() {
		return catalog;
	}

	public void setCatalog(String catalog) {
		this.catalog = catalog;
	}

	public BlockingQueue<MessageEntity> getQueue() {
		return queue;
	}

	public void setQueue(BlockingQueue<MessageEntity> queue) {
		this.queue = queue;
	}

}